package com.sc.nft.admin.consume;

import com.alibaba.fastjson.JSONObject;
import com.sc.nft.dto.TradePoolConsumerDTO;
import com.sc.nft.enums.MessageTopic;
import com.sc.nft.enums.RedisKeyEnum;
import com.sc.nft.exception.GlobalRunTimeException;
import com.sc.nft.service.IncomeService;
import com.sc.nft.sup.ErrorCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@RocketMQMessageListener(topic = MessageTopic.TRADE_POOL_INCOME + "${spring.profiles.active}", consumerGroup = MessageTopic.TRADE_POOL_INCOME + "${spring.profiles.active}")
@Component
@Slf4j
public class TradePoolConsumer implements RocketMQListener<TradePoolConsumerDTO> {

    @Resource
    private Redisson redisson;

    @Resource
    private IncomeService incomeService;

    @Override
    public void onMessage(TradePoolConsumerDTO transPool) {
        log.info("交易池收益分润收到消息:{}", JSONObject.toJSONString(transPool));
        RLock lock = redisson.getLock(RedisKeyEnum.TRADE_POOL_LOCK.getKey() + transPool.getTradeOrderId());
        try {
            boolean res = lock.tryLock(10, TimeUnit.SECONDS);
            if (res) {
                log.info("交易池收益:{}开始结算", transPool.getTradeOrderId());
                incomeService.tradePoolIncome(transPool);
            } else {
                throw new GlobalRunTimeException(ErrorCode.FIRSTLEVE_INCOME_CAL_ERROR);
            }
        } catch (Exception e) {
            log.error("一级收益消费异常:{}", transPool, e);
            throw new GlobalRunTimeException(ErrorCode.FIRSTLEVE_INCOME_CAL_ERROR);
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        log.info("交易池收益分润结算完成:{}", JSONObject.toJSONString(transPool));
    }
}
